package com.yxx.servicealbum.service.impl;

import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.baomidou.mybatisplus.core.metadata.IPage;
import com.baomidou.mybatisplus.core.toolkit.CollectionUtils;
import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import com.yxx.kafkautil.constant.KafkaConstant;
import com.yxx.kafkautil.service.KafkaService;
import com.yxx.model.album.AlbumAttributeValue;
import com.yxx.model.album.AlbumInfo;
import com.yxx.model.album.AlbumStat;
import com.yxx.query.album.AlbumInfoQuery;
import com.yxx.result.Result;
import com.yxx.serverutil.common.YxxRedissonCache;
import com.yxx.serverutil.config.redisson.RedissonConfig;
import com.yxx.serverutil.constant.RedisConstant;
import com.yxx.serverutil.constant.SystemConstant;
import com.yxx.servicealbum.mapper.AlbumAttributeValueMapper;
import com.yxx.servicealbum.mapper.AlbumInfoMapper;
import com.yxx.servicealbum.mapper.AlbumStatMapper;
import com.yxx.servicealbum.service.AlbumAttributeValueService;
import com.yxx.servicealbum.service.AlbumInfoService;
import com.yxx.vo.album.AlbumAttributeValueVo;
import com.yxx.vo.album.AlbumInfoVo;
import com.yxx.vo.album.AlbumListVo;
import com.yxx.vo.album.AlbumStatVo;
import lombok.SneakyThrows;
import org.redisson.api.RBloomFilter;
import org.redisson.api.RLock;
import org.redisson.api.RedissonClient;
import org.springframework.beans.BeanUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.core.script.DefaultRedisScript;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.PathVariable;

import java.util.Arrays;
import java.util.List;
import java.util.UUID;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;

/**
 * @author yuxiangxun
 * @date 2025/3/16
 * @apiNote
 */
@Service
@SuppressWarnings({"unchecked", "rawtypes"})
public class AlbumInfoServiceImpl extends ServiceImpl<AlbumInfoMapper, AlbumInfo> implements AlbumInfoService {

    @Autowired
    private AlbumAttributeValueService albumAttributeValueService;

    @Autowired
    private AlbumStatMapper albumStatMapper;

    @Autowired
    private AlbumInfoMapper albumInfoMapper;

    @Autowired
    private AlbumAttributeValueMapper albumAttributeValueMapper;

    @Autowired
    private KafkaService kafkaService;

    @Autowired
    private RedisTemplate redisTemplate;

    @Autowired
    private RedissonClient redissonClient;



    @Override
    @Transactional(rollbackFor = Exception.class)
    public void saveAlbumInfo(AlbumInfoVo albumInfoVo, Long userId) {
        AlbumInfo albumInfo = new AlbumInfo();
        BeanUtils.copyProperties(albumInfoVo,albumInfo);
        // 设置保存状态审核通过
        albumInfo.setStatus(SystemConstant.ALBUM_STATUS_PASS);
        // 设置userId
        albumInfo.setUserId(userId);
        // 判断这个专辑是否免费吧，不免费则设置前五集免费
        if(!SystemConstant.ALBUM_PAY_TYPE_FREE.equals(albumInfoVo.getPayType())){
            albumInfo.setTracksForFree(5);
        }
        this.save(albumInfo);

        // 批量保存专辑属性标签  (还要判断这里的albumAttributeValueVoList是否为空)
        List<AlbumAttributeValueVo> albumAttributeValueVoList = albumInfoVo.getAlbumAttributeValueVoList();
        if(CollectionUtils.isNotEmpty(albumAttributeValueVoList)){
            List<AlbumAttributeValue> albumAttributeValueList = albumAttributeValueVoList.stream().map(albumAttributeValueVo -> {
                AlbumAttributeValue albumAttributeValue = new AlbumAttributeValue();
                BeanUtils.copyProperties(albumAttributeValueVo, albumAttributeValue);
                albumAttributeValue.setAlbumId(albumInfo.getId());
                albumAttributeValue.setAttributeName(albumInfo.getAlbumTitle());
                return albumAttributeValue;
            }).collect(Collectors.toList());
            // 批量保存专辑属性标签数据
            this.albumAttributeValueService.saveBatch(albumAttributeValueList);
        }


        // 保存专辑统计数据
        this.saveAlbumStat(albumInfo.getId(), SystemConstant.ALBUM_STAT_BUY);
        this.saveAlbumStat(albumInfo.getId(), SystemConstant.ALBUM_STAT_PLAY);
        this.saveAlbumStat(albumInfo.getId(), SystemConstant.ALBUM_STAT_SUBSCRIBE);
        this.saveAlbumStat(albumInfo.getId(), SystemConstant.ALBUM_STAT_COMMENT);

        // 保存专辑完成后，将专辑id保存到布隆过滤器
        RBloomFilter<Object> bloomFilter = this.redissonClient.getBloomFilter(RedisConstant.ALBUM_BLOOM_FILTER);
        bloomFilter.add(albumInfo.getId());

    }

    @Override
    public IPage<AlbumListVo> findUserAlbumPage(Page<AlbumListVo> albumInfoPage, AlbumInfoQuery albumInfoQuery) {
        return albumInfoMapper.selectUserAlbumPage(albumInfoPage,albumInfoQuery);
    }

    @Override
    @Transactional(rollbackFor = Exception.class)
    public void removeAlbumInfoById(Long id) {
        // 根据专辑id删除专辑信息
        this.removeById(id);
        // 删除专辑对应的属性标签
        albumAttributeValueMapper.delete(new LambdaQueryWrapper<AlbumAttributeValue>().eq(AlbumAttributeValue::getAlbumId,id));
        // 删除专辑对应的统计数据
        albumStatMapper.delete(new LambdaQueryWrapper<AlbumStat>().eq(AlbumStat::getAlbumId,id));
        // 删除时专辑自动下架
        this.kafkaService.sendMessage(KafkaConstant.QUEUE_ALBUM_LOWER,String.valueOf(id));
    }

    /**
     * 回显专辑数据
     * 法一：使用 Redis 分布式锁
     * 法二：使用 Redisson 整合分布式锁
     * @param id
     * @return
     */
    @YxxRedissonCache(prefix = RedisConstant.ALBUM_INFO_PREFIX)
    @Override
    public AlbumInfo getAlbumInfoById(Long id) {
        AlbumInfo albumInfo = this.getById(id);
        // 还需要回显专辑属性数据信息数据
        List<AlbumAttributeValue> albumAttributeValues = albumAttributeValueMapper.selectList(new LambdaQueryWrapper<AlbumAttributeValue>()
                .eq(AlbumAttributeValue::getAlbumId, id));
        albumInfo.setAlbumAttributeValueVoList(albumAttributeValues);
        return albumInfo;
    }

    //@Override
    //public AlbumInfo getAlbumInfoById(Long id){
    //    // 使用 Redisson 整合分布式锁
    //    // 创建 AlbumInfo 对象
    //    AlbumInfo albumInfo = new AlbumInfo();
    //    // 声明 缓存 key -> album:info:id
    //    String key = RedisConstant.ALBUM_INFO_PREFIX + id;
    //    try{
    //        // 通过缓存 key 从 Redis 中获取数据
    //        albumInfo = (AlbumInfo) redisTemplate.opsForValue().get(key);
    //        // 如果从缓存中获取不到数据
    //        if(null == albumInfo){
    //            // 声明专辑锁的key -> album:lock:id:lock
    //            String albumInfoLockKey = RedisConstant.ALBUM_LOCK_PREFIX + id + ":lock";
    //            // 从 Redisson 中获取专辑锁对象
    //            RLock lock = redissonClient.getLock(albumInfoLockKey);
    //            // Redisson 上锁操作
    //            Boolean isSuccessLock = lock.tryLock(RedisConstant.CACHE_LOCK_EXPIRE_PX1,RedisConstant.CACHE_LOCK_EXPIRE_PX2,TimeUnit.SECONDS);
    //            if(isSuccessLock){
    //                try{
    //                    // 上锁成功
    //                    // 查数据库
    //                    albumInfo = this.getById(id);
    //                    // 查数据库对应该专辑数据为空
    //                    if(null == albumInfo){
    //                        AlbumInfo albumInfoToNUll = new AlbumInfo();
    //                        this.redisTemplate.opsForValue().set(key,albumInfoToNUll,RedisConstant.ALBUM_TEMPORARY_TIMEOUT,TimeUnit.SECONDS);
    //                        return albumInfoToNUll;
    //                    }
    //                    // 查数据库对应专辑数据不为空，缓存该专辑数据并返回
    //                    this.redisTemplate.opsForValue().set(key,albumInfo,RedisConstant.ALBUM_TIMEOUT,TimeUnit.SECONDS);
    //                    return albumInfo;
    //                }finally {
    //                    // 解锁
    //                    lock.unlock();
    //                }
    //            }else{
    //                // 上锁失败，说明有线程占用该锁，其他线程应该自旋
    //                return getAlbumInfoById(id);
    //            }
    //
    //
    //        }else{
    //            // 从缓存中可以获取到数据，直接返回
    //            return albumInfo;
    //        }
    //    }catch(Exception e){
    //        e.printStackTrace();
    //    }
    //    // 出现异常了（如Redis宕机了），这里是保险措施，直接查询数据库中的数据返回
    //    return this.getById(id);
    //}

    //@Override
    //public AlbumInfo getAlbumInfoById(Long id){
    //    // 使用 Redis 分布式锁整合业务
    //    // 总体思路：从缓存中查不到数据，只需上锁，通过一个线程（请求）进行查询操作并将数据缓存起来，这些业务操作成功后解锁，其他线程（请求）自旋->就先查缓存有无数据，有的话直接获取返回了
    //    // 创建专辑对象
    //    AlbumInfo albumInfo = new AlbumInfo();
    //    // 声明 缓存 key -> album:info:id
    //    String key = RedisConstant.ALBUM_INFO_PREFIX + id;
    //    try{
    //        // 通过缓存 key 从 Redis 中获取数据
    //        albumInfo =  (AlbumInfo) redisTemplate.opsForValue().get(key);
    //        // 如果从缓存中获取到的数据集为空
    //        if(null == albumInfo){
    //            // 声明专辑锁的key -> album:lock:id:lock
    //            String albumLockKey = RedisConstant.ALBUM_LOCK_PREFIX + id + ":lock";
    //            // 给这把锁上唯一标识，防止高并发时其他线程误删锁操作
    //            String uuid = UUID.randomUUID().toString();
    //            // 上锁操作
    //            Boolean isSuccessLock = redisTemplate.opsForValue().setIfAbsent(albumLockKey, uuid, RedisConstant.ALBUM_LOCK_EXPIRE_PX2, TimeUnit.SECONDS);
    //            if(isSuccessLock){
    //               try{
    //                   // 上锁成功，且该线程拿到锁，则进行业务操作
    //                   // 从数据库中查询数据
    //                   albumInfo = this.getById(id);
    //                   // 如果从数据库中查询不到该数据
    //                   if(null == albumInfo){
    //                       // 创建空对象，缓存空对象
    //                       AlbumInfo albumInfoToNull = new AlbumInfo();
    //                       redisTemplate.opsForValue().set(key,albumInfoToNull,RedisConstant.ALBUM_TEMPORARY_TIMEOUT,TimeUnit.SECONDS);
    //                       return albumInfoToNull;
    //                   }
    //                   // 从数据库中成功查询到该专辑信息，进行缓存并返回
    //                   redisTemplate.opsForValue().set(key,albumInfo,RedisConstant.ALBUM_TIMEOUT,TimeUnit.SECONDS);
    //                   return albumInfo;
    //               }finally {
    //                   // 执行完业务操作，解锁，使用 lua 脚本进行解锁（防止误删其他线程的锁）
    //                   delRedisKey(albumLockKey,uuid);
    //               }
    //            }else{
    //                // 有线程占有该锁，本线程应该等待其释放锁（自旋）
    //                return getAlbumInfoById(id);
    //            }
    //        }else{
    //            // 查询缓存不为空，直接返回缓存中的数据
    //            return albumInfo;
    //        }
    //    }catch(Exception e){
    //        e.printStackTrace();
    //    }
    //    // 出现异常了（如Redis宕机了），这里是保险措施，直接查询数据库中的数据返回
    //    return this.getById(id);
    //}


    /**
     * 删除缓存的key
     * @param albumLockKey
     * @param uuid
     */
    public void delRedisKey(String albumLockKey, String uuid){
        //4.1 先创建脚本对象 DefaultRedisScript泛型脚本语言返回值类型 Long 0：失败 1：成功
        DefaultRedisScript defaultRedisScript = new DefaultRedisScript<>();
        //4.2设置脚本文本
        String scriptText = "if redis.call(\"get\",KEYS[1]) == ARGV[1]\n" +
                "then\n" +
                "    return redis.call(\"del\",KEYS[1])\n" +
                "else\n" +
                "    return 0\n" +
                "end";
        defaultRedisScript.setScriptText(scriptText);
        //4.3 设置响应类型
        defaultRedisScript.setResultType(Long.class);
        redisTemplate.execute(defaultRedisScript, Arrays.asList(albumLockKey),uuid);
    }


    /**
     * 更新专辑信息
     * 因为上了缓存，为了保证数据一致性，使用延迟双删来保证缓存与数据库数据一致性！
     * @param id
     * @param albumInfoVo
     */
    @Override
    @Transactional(rollbackFor = Exception.class)
    public void updateAlbumInfo(Long id, AlbumInfoVo albumInfoVo) {
        // 更新策略：延迟双删  —— 解决MySQL与Redis的数据一致性问题
        // 先构建缓存key   album:info:[专辑id]
        // 先删缓存
        String key = RedisConstant.ALBUM_INFO_PREFIX + "[" + id + "]";
        redisTemplate.delete(key);

        // 通过传递进来的专辑id，向数据库发起查询，获得旧专辑信息对象
        // 对象拷贝  ->  新专辑数据 替换掉  旧专辑数据
        // 更新专辑数据
        AlbumInfo albumInfo = this.getById(id);
        BeanUtils.copyProperties(albumInfoVo,albumInfo);
        this.updateById(albumInfo);


        // 删除旧的专辑属性数据
        // 批量保存新的专辑属性数据
        albumAttributeValueMapper.delete(new LambdaQueryWrapper<AlbumAttributeValue>().eq(AlbumAttributeValue::getAlbumId,id));
        List<AlbumAttributeValueVo> albumAttributeValueVoList = albumInfoVo.getAlbumAttributeValueVoList();
        if(CollectionUtils.isNotEmpty(albumAttributeValueVoList)){
            List<AlbumAttributeValue> albumAttributeValueList = albumAttributeValueVoList.stream().map(item -> {
                AlbumAttributeValue albumAttributeValue = new AlbumAttributeValue();
                BeanUtils.copyProperties(item, albumAttributeValue);
                albumAttributeValue.setAlbumId(id);
                return albumAttributeValue;
            }).collect(Collectors.toList());
            albumAttributeValueService.saveBatch(albumAttributeValueList);
        }


        // 阻塞10秒，让并发先更新Redis，因为并发访问，访问的大概率还是旧数据，阻塞10s后，再删除Redis缓存
        try{
            Thread.sleep(10);
        }catch (Exception e){
            throw new RuntimeException(e);
        }
        redisTemplate.delete(key);

        // 有个公开和私密的功能，公开表示专辑上架，私密表示专辑下架
        // 这里需要判断是否公开，公开就发消息让专辑上架服务接收执行上架操作，并且将更新后的专辑id加入布隆过滤器中，私密就发消息让给专辑下架服务接收执行下架操作
        if("1".equals(albumInfoVo.getIsOpen())){
            RBloomFilter<Object> bloomFilter = redissonClient.getBloomFilter(RedisConstant.ALBUM_BLOOM_FILTER);
            bloomFilter.add(albumInfo.getId());
            kafkaService.sendMessage(KafkaConstant.QUEUE_ALBUM_LOWER, String.valueOf(albumInfo.getId()));
        }else{
            kafkaService.sendMessage(KafkaConstant.QUEUE_ALBUM_LOWER, String.valueOf(albumInfo.getId()));
        }
    }

    @Override
    public List<AlbumInfo> findUserAllAlbumList(Long userId) {
        // 默认查看第一页
        Page<AlbumInfo> albumInfoPage = new Page<>(1, 100);
        // 设置查询条件
        LambdaQueryWrapper<AlbumInfo> albumInfoLambdaQueryWrapper = new LambdaQueryWrapper<AlbumInfo>().select(AlbumInfo::getId, AlbumInfo::getAlbumTitle)
                .eq(AlbumInfo::getUserId, userId)
                .orderByDesc(AlbumInfo::getId);
        // 返回查询记录
        return albumInfoMapper.selectPage(albumInfoPage,albumInfoLambdaQueryWrapper).getRecords();
    }

    @Override
    public List<AlbumAttributeValue> findAlbumAttributeValueByAlbumId(Long albumId) {
        List<AlbumAttributeValue> albumAttributeValueList = albumAttributeValueMapper.selectList(new LambdaQueryWrapper<AlbumAttributeValue>()
                .eq(AlbumAttributeValue::getAlbumId, albumId));
        return albumAttributeValueList;
    }

    @Override
    @YxxRedissonCache(prefix = RedisConstant.ALBUM_STAT_PREFIX)
    public AlbumStatVo getAlbumStatVoByAlbumId(Long albumId) {
        //	调用mapper 层方法
        return albumInfoMapper.selectAlbumStat(albumId);
    }

    @Override
    public void updateStat(Long albumId, String statType, Integer count) {
        albumInfoMapper.updateStat(albumId,statType,count);
    }

    /**
     * 初始化统计数据
     * @param albumId
     * @param statType
     */
    private void saveAlbumStat(Long albumId, String statType){
        AlbumStat albumStat = new AlbumStat();
        albumStat.setAlbumId(albumId);
        albumStat.setStatType(statType);
        albumStat.setStatNum(0);
        albumStatMapper.insert(albumStat);
    }


}
